package com.hp.provider;

import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import com.hp.provider.api.RocketMQOtherTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

@SpringBootApplication
//@EnableBinding({PropertiesSource.class})
@RestController
public class SpringRocketmqProviderApplication implements CommandLineRunner {

    //@Autowired
    //private PropertiesSource propertiesSource;

    @Autowired
    private RocketMQOtherTemplate rocketMQTemplate;

    @Resource
    private StreamBridge streamBridge;

    public static void main(String[] args) {
        SpringApplication.run(SpringRocketmqProviderApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {

        /**
         * 使用原生的Rocketmq的发送消息
         */
//        org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message("test-topic", null, "123456", "哈哈哈".getBytes());
////        message.setDelayTimeLevel(0);
//        rocketMQTemplate.getProducer().send(message);


//        while (true) {
//            Thread.sleep(2000);
//            for (int i = 0; i < 10; i++) {
//                sendWithKeys("王五"+i,i+"");
//                sendWithTags("田七"+i,i+"");
//            }
//            break;
//        }
    }

    @GetMapping("/test")
    public void test() throws Exception {
//        org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message("test-topic", null, "123456", "哈哈哈".getBytes());
////        message.setDelayTimeLevel(0);
//        try {
//            rocketMQTemplate.getProducer().send(message);
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        } catch (RemotingException e) {
//            e.printStackTrace();
//        } catch (MQBrokerException e) {
//            e.printStackTrace();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        for (int i = 0; i < 10; i++) {
            //sendWithKeys("王五" + i, i + "");
            //sendWithTags("田七" + i, i + "");
            //sendWithKeysAndTags1("测试 topic-1-" + i, i + "tag", i + "key");
            //sendWithKeysAndTags2("测试 topic-2-" + i, i + "tag", i + "key");

            sendWithKeysAndTags1(User.builder().name("rocket").age(i).build(),i + "tag", i + "key");
            sendWithKeysAndTags2(User.builder().name("rocket").age(i).build(),i + "tag", i + "key");
            sendWithKeysAndTags3(User.builder().name("kafka").age(i).build(),i + "tag", i + "key");
            sendWithKeysAndTags4(User.builder().name("kafka").age(i).build(),i + "tag", i + "key");

        }
    }

    //public void send(String msg) throws Exception {
    //    propertiesSource.output1().send(MessageBuilder.withPayload(msg).build());
    //}
    //
    //public <T> void sendWithTags(T msg, String tag) throws Exception {
    //    Message message = MessageBuilder.createMessage(msg,
    //            new MessageHeaders(Stream.of(tag).collect(Collectors
    //                    .toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
    //    propertiesSource.output1().send(message);
    //}
    //
    //public <T> void sendWithKeys(T msg, String key) throws Exception {
    //    Message message = MessageBuilder.createMessage(msg,
    //            new MessageHeaders(Stream.of(key).collect(Collectors
    //                    .toMap(str -> MessageConst.PROPERTY_KEYS, String::toString))));
    //    propertiesSource.output1().send(message);
    //}
    //
    //public <T> void sendObject(T msg, String tag) throws Exception {
    //    Message message = MessageBuilder.withPayload(msg)
    //            .setHeader(MessageConst.PROPERTY_TAGS, tag)
    //            .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
    //            .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 1)
    //            .build();
    //    propertiesSource.output1().send(message);
    //}

    public <T> void sendWithKeysAndTags1(T msg, String tag, String key) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQConst.Headers.TAGS, tag)
                .setHeader(RocketMQConst.Headers.KEYS, key)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                //.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 1)
                .build();
        streamBridge.send("outputRocket1-out-0",message);
    }


    public <T> void sendWithKeysAndTags2(T msg, String tag, String key) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQConst.Headers.TAGS, tag)
                .setHeader(RocketMQConst.Headers.KEYS, key)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                //.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 1)
                .build();
        streamBridge.send("outputRocket2-out-0",message);
    }


    public <T> void sendWithKeysAndTags3(T msg, String tag, String key) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQConst.Headers.TAGS, tag)
                .setHeader(RocketMQConst.Headers.KEYS, key)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                //.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 1)
                .build();
        streamBridge.send("outputKafka1-out-0",message);
    }


    public <T> void sendWithKeysAndTags4(T msg, String tag, String key) throws Exception {
        Message message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQConst.Headers.TAGS, tag)
                .setHeader(RocketMQConst.Headers.KEYS, key)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                //.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 1)
                .build();
        streamBridge.send("outputKafka2-out-0",message);
    }


//    public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
//        MessageBuilder builder = MessageBuilder.withPayload(msg)
//                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
//        builder.setHeader("test", String.valueOf(num));
//        builder.setHeader(RocketMQHeaders.TAGS, "binder");
//        Message message = builder.build();
//        propertiesSource.output2().send(message);
//    }

}
